package cn.com.cennavi.HighprecisionTrafficShow;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.regex.Pattern;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;

/**
 * 轨迹数据筛选
 * 
 * @ClassName: DataScreen
 * @Description: TODO（这里用一句话描述这个类的作用）
 * @author bull
 * @date 2016年3月15日 下午2:04:20
 *
 */
public class DataScreen {
	private static final Pattern COMMA = Pattern.compile(",");

	public static void main(String[] args) {
		// TODO Auto-generated method stub

		// 配置spark
		String trajFilePaths = "file:///D:\\data\\BASIS_TRAJECTORY_2016_0";
		String differPointMapFilePath = "file:///C:\\Users\\NXZ\\Desktop\\lonlat.csv";
		SparkConf conf = new SparkConf().setAppName("testLocal").setMaster("local");
		JavaSparkContext sc = new JavaSparkContext(conf);

		// 读取分歧点地图数据areaCodeLinkSeqMap：key：110000，value：hashmap(linkseq,mapStr)

		JavaPairRDD<String, String> difPointMapData = sc.textFile(differPointMapFilePath).cache().mapToPair(new PairFunction<String, String, String>() {

			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<String, String> call(String line) throws Exception {
				// TODO Auto-generated method stub
				String areaCode = "110000";
				return new Tuple2<String, String>(areaCode, line);
			}
		});
		JavaPairRDD<String, HashMap<String, String>> areaCodeLinkSeqMap = difPointMapData.groupByKey().mapToPair(new PairFunction<Tuple2<String, Iterable<String>>, String, HashMap<String, String>>() {

			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<String, HashMap<String, String>> call(Tuple2<String, Iterable<String>> mapData) throws Exception {
				// TODO Auto-generated method stub
				String areaCode = mapData._1;
				HashMap<String, String> linkseqMap = new HashMap<>();
				for (String line : mapData._2) {
					String[] arr = COMMA.split(line);
					String lineseq = arr[1];
					linkseqMap.put(lineseq, "1");
				}
				return new Tuple2<String, HashMap<String, String>>(areaCode, linkseqMap);
			}
		});

		// 读取轨迹数据：key：100010，value：TrajectoryString
		JavaPairRDD<String, Tuple2<String, HashMap<String, String>>> trajectoryData = sc.textFile(trajFilePaths).cache().mapToPair(new PairFunction<String, String, String>() {

			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<String, String> call(String line) throws Exception {
				// TODO Auto-generated method stub
				String areaCode = "110000";
				return new Tuple2<String, String>(areaCode, line);
			}
		}).join(areaCodeLinkSeqMap).cache();

		//
		JavaPairRDD<String, ArrayList<String>> screenData = trajectoryData.mapToPair(new PairFunction<Tuple2<String, Tuple2<String, HashMap<String, String>>>, String, ArrayList<String>>() {

			private static final long serialVersionUID = 1L;

			@Override
			public Tuple2<String, ArrayList<String>> call(Tuple2<String, Tuple2<String, HashMap<String, String>>> data) throws Exception {
				// TODO Auto-generated method stub
				String areaCode = data._1;
				String trajectoryData = data._2._1;
				HashMap<String, String> linkMap = data._2._2;
				String[] dataArr = trajectoryData.split(",");
				ArrayList<String> datalist = getLinkSeqList(trajectoryData, linkMap);
				return new Tuple2<String, ArrayList<String>>(areaCode, datalist);
			}

			/**
			 * 一条轨迹中包含的link序列属于分歧点道路不？
			 */
			private ArrayList<String> getLinkSeqList(String line, HashMap<String, String> linkMap) {
				// TODO Auto-generated method stub
				ArrayList<String> dataList = new ArrayList<>();
				String[] lineArr = line.split(",");
				if (lineArr.length != 30) {
					return null;
				} else {
					String rowkey = lineArr[0];
					String linkSequence = lineArr[4];
					String orgGPSSequence = lineArr[28];
					String enterTimeSeq = lineArr[8];
					for (Map.Entry<String, String> entry : linkMap.entrySet()) {
						String linkSeq = entry.getKey();
						if (linkSequence.contains(linkSeq)) {
							/** 说明包含一条分歧点道路，接下来求取落在分歧点道路里面的gps原始点信息 **/
							String goodGPSInfo = getGoodTimePair(linkSequence, linkSeq, enterTimeSeq, orgGPSSequence);
							String line1 = rowkey + "," + goodGPSInfo + "," + linkSeq;
							dataList.add(line1);
						}
					}
				}
				return dataList;
			}

			/**
			 * 得到落在link序列内的所有gos原始点信息
			 */
			private String getGoodTimePair(String linkSequence, String linkSeq, String enterTimeSeq, String orgGPSSequence) {
				// TODO Auto-generated method stub
				/** 获取link序列的进入时间和结束时间 **/
				String[] enterTimeSeqArr = enterTimeSeq.split("\\|");
				String[] linkArr = linkSeq.split("\\|");
				String[] linkSequenceArr = linkSequence.split("\\|");
				int size = linkArr.length;
				String oLink = linkArr[0];
				int oIndex = 0;
				for (int i = 0; i < linkSequenceArr.length; i++) {
					if (linkSequenceArr[i].equals(oLink)) {
						oIndex = i;
						break;
					}
				}
				int totalIndex = oIndex + size;
				int dIndex = totalIndex -(linkSequenceArr.length-1) > 0 ? linkSequenceArr.length-1 : totalIndex;
				String[] enterTimeArr = enterTimeSeq.split("\\|");
				long firstTime = Long.parseLong(enterTimeArr[oIndex]);
				long lastTime = Long.parseLong(enterTimeArr[dIndex]);
				
				/**
				 * 已经得到了link序列的进入时间和结束时间，循环便利gps原始点序列，如果时间戳落在进入时间和结束时间之间，
				 * 则认为是要提取的点
				 **/
				StringBuffer sb = new StringBuffer();
				String[] gpsArr = orgGPSSequence.split("\\|");
				
				String []arr0=gpsArr[0].split("\\:");
				long lon=Long.parseLong(arr0[0]);
				long lat=Long.parseLong(arr0[1]);
				long timeStamp=Long.parseLong(arr0[3]);
				for (int j=0; j<gpsArr.length;j++) {
					String []arr=gpsArr[j].split("\\:");
					if(j==0){
						if (timeStamp < firstTime || timeStamp > lastTime) {
							continue;
						} else {
							sb.append(gpsArr[j]);
							sb.append("|");
						}
					}else{
						long nowTime=timeStamp+Long.parseLong(arr[3]);
						long nowLon=lon+Long.parseLong(arr[0]);
						long nowLat=lat+Long.parseLong(arr[1]);
						int nowSpeed=Integer.parseInt(arr[2]);
						int dir=Integer.parseInt(arr[4]);
						if (nowTime < firstTime || nowTime > lastTime) {
							continue;
						} else {
							sb.append(nowLon+":"+nowLat+":"+nowSpeed+":"+nowTime+":"+dir);
							sb.append("|");
						}
					}
					
				}
//				System.out.println(sb.toString());
				return sb.toString();
			}
		});
		List<Tuple2<String, ArrayList<String>>> output = screenData.collect();
		for (Tuple2<String, ArrayList<String>> tuple : output) {
			ArrayList<String> dataList = tuple._2;
			if (dataList == null) {
				continue;
			}
//			for (String s : dataList) {
//				System.out.println(tuple._1() + ": " + s);
//			}

		}
		sc.stop();
	}

}
